package demo.kafka.apache;

import demo.utils.Constants;
import demo.utils.ThreadUtil;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.clients.producer.internals.DefaultPartitioner;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.errors.AuthorizationException;
import org.apache.kafka.common.errors.OutOfOrderSequenceException;
import org.apache.kafka.common.errors.ProducerFencedException;
import org.junit.Before;
import org.junit.Test;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;


/**
 * <h1>发送的消息如何选择分区：</h1>
 * <li>如果我们创建ProducerRecord时设置了partition，就用我们设置的。否则进行计算。
 * <li>默认的实现是DefaultPartitioner:如果keyBytes为null 就使用轮询选择分区。否则根据key的hash值。
 *
 */
public class ApacheProducerDemo {
    DefaultPartitioner a = null;

    public static final String TOPIC = "test.stan";

    private Properties properties;

    @Before
    public void setUp() {
        properties = new Properties();
        // Kafka服务端的主机名和端口号
        properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, Constants.KAFKA_BOOTSTRAP_SERVERS);
        // 客户的ID
        properties.put(CommonClientConfigs.CLIENT_ID_CONFIG, "ProducerDemo");
        // 消息的key和value都是字节数组，为了将Java对象转化为字节数组，可以配置
        // key.serializer和value.serializer两个序列化器，完成转化
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");
        // StringSerializer用来将String对象序列化成字节数组
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");

        // -------------  事务相关配置 ------------------
//        initTransactionCfg(properties);
        // 可选的 ACKS 配置，通常使用 ‘all‘ 以确保最高可靠性
        properties.put(ProducerConfig.ACKS_CONFIG, "all");

        // ----------------- 指定拦截器 ------------------
        // 多个拦截器：用逗号分隔，会按顺序执行
//        properties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, "demo.mq.kafka.interceptor.InfraKafkaProducerInterceptor");
    }

    /**
     * 初始化生产者配置
     *
     * @return
     */
    public static Properties initProducerProperty() {
        ApacheProducerDemo demo = new ApacheProducerDemo();
        demo.setUp();
        return demo.properties;
    }

    /**
     * 初始化事务配置
     *
     * @param properties
     */
    public static void initTransactionCfg(Properties properties) {
        // 唯一的事务ID，用于故障恢复
        properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "test-transactional-id");
        // 启用幂等性是使用事务的前提
        properties.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
    }

    /**
     * 同步发送消息。
     * KafkaProducer.send()方法的返回值类型是Future<RecordMetadata>，
     * 这里通过Future.get()方法，阻塞当前线程，等待Kafka服务端的ACK响应.
     * <p>
     * 缺点：
     * <li>性能极差：每次发送都会引入网络往返的延迟，吞吐量会急剧下降。通常不推荐在生产环境中大量使用。
     */
    @Test
    public void testSendSync() throws ExecutionException, InterruptedException {
        // 生产者核心类
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

        String msgKey = "key1";
        String msgValue = "hello kafka! 了却君王天下事赢得生前身后名！";
        ProducerRecord<String, String> producerRecord = new ProducerRecord<>(TOPIC, msgKey, msgValue);
        Future<RecordMetadata> future = kafkaProducer.send(producerRecord);
        RecordMetadata recordMetadata = future.get();
        // 消息发送成功！
        System.err.println("消息发送成功至: Topic - " + recordMetadata.topic() +
                ", Partition：" + recordMetadata.partition() +
                ", Offset： " + recordMetadata.offset());
        kafkaProducer.close();
    }

    /**
     * <h1>异步发送 + 回调（Callback）（推荐）</h1>
     * 最常用且性能与可靠性兼备的方式。你传入一个回调函数，当发送成功或失败时，Sender 线程会异步地调用这个回调函数进行通知，主线程不会被阻塞。
     * CallBack对象，当生产者接收到Kafka发来的ACK确认消息的时候，会调用此CallBack对象的onCompletion()方法，实现 回调功能
     * <p>
     * 优点：
     *
     * <li>高吞吐量：消息被放入缓冲区后立即返回，不会阻塞。
     * <li>高可靠性：能明确知道每条消息的最终状态，便于错误处理和重试。
     * <li>获取元数据：成功时可以拿到消息的详细位置信息。
     */
    @Test
    public void testSendAsync() {
        // 生产者核心类
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
        String msgKey = "key2";
        String msgValue = "hello kafka!！！";
        long startTime = System.currentTimeMillis();

        ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, msgKey, msgValue);
        kafkaProducer.send(record, new DemoCallBack<>(startTime, msgKey, msgValue));

        ThreadUtil.safeSleep(3000L);
        kafkaProducer.close();
    }


    /**
     * 自定义的回调函数，用于处理KafkaProducer.send()方法的返回值
     *
     * @param <K>
     * @param <V>
     */
    public static class DemoCallBack<K, V> implements Callback {
        private final long startTime;
        private final K key;
        private final V value;

        public DemoCallBack(long startTime, K key, V value) {
            this.startTime = startTime;
            this.key = key;
            this.value = value;
        }

        /**
         * 生产者成功发送消息，收到Kafka服务端发来的ACK确认消息后，会调用此回调函数
         *
         * @param recordMetadata 生产者发送的消息的元数据，如果发送过程中出现异常，此参数为null
         * @param exception      发送过程中出现的异常，如果发送成功，则此参数为null
         */
        @Override
        public void onCompletion(RecordMetadata recordMetadata, Exception exception) {
            if (recordMetadata != null) {
                long elapsedTime = System.currentTimeMillis() - startTime;
                String msg = String.format("message(%s) send to partition(%s),offset(%s) ,cost %s ms",
                        key, recordMetadata.partition(), recordMetadata.offset(), elapsedTime);
                System.err.println(msg);
            } else {
                exception.printStackTrace();
            }
        }

    }


    /**
     * 详细介绍一下 Kafka 在发送消息时的事务（Transaction）功能。这是一个非常重要的特性，它解决了“精确一次”（Exactly-Once）语义在生产者端的问题。
     * <p>
     * 在没有事务之前，Kafka 生产者可能会遇到以下问题：
     * <p>
     * <li>重复消息：如果生产者发送消息后，网络抖动导致 Broker 的确认（ACK）没有成功返回，生产者会认为消息发送失败并进行重试。这就会导致同一条消息被 Broker 存储多次。
     * <li>消息错乱：在某些需要保证消息顺序的业务场景中，如果重试了多条消息，可能会破坏消息的先后顺序。
     * <li>跨分区、跨主题的原子性写入失败：一个业务操作可能需要向多个 Topic 或 Partition 发送消息。在非事务情况下，如果部分消息发送成功，部分失败，就会造成数据不一致。
     * <p>
     * 事务的核心目的就是解决上述问题，保证多个消息的发送（可能涉及多个分区和主题） 在一个原子操作中：要么全部成功，要么全部失败。
     *
     * <h2>事务的核心思想</h2>
     * Kafka 事务的实现借鉴了数据库的事务模型，其核心思想是：
     * <li>原子性提交： 将一批消息的发送包装成一个原子操作。
     * <li>隔离性： 在事务提交之前，事务内部的消息对消费者是不可见的（通过 isolation.level 参数控制）。
     * <li>持久性： 一旦事务提交，消息就被持久化，即使 Broker 重启也不会丢失。
     *
     *
     * <h2>消费者的隔离级别</h2>
     * 事务不仅关乎生产者，消费者也需要进行相应配置来决定如何读取事务消息。要实现完整的事务语义，生产者和消费者必须配合使用。
     * 消费者通过 isolation.level 参数来控制：
     * <li>read_uncommitted（默认）： 读取所有消息，无论事务是提交还是中止。这意味着消费者可能会读到未提交（最终被回滚） 的消息。
     * <li>read_committed： 只读取已提交事务的消息。对于未提交事务的消息或正在进行中的事务消息，消费者会等待直到事务提交或中止后，才会看到最终结果。这保证了消费者不会看到中间状态或脏数据。
     */
    @Test
    public void testTransaction() {
        // 初始化事务配置
        initTransactionCfg(properties);
        // 生产者核心类
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
        // 初始化事务
        kafkaProducer.initTransactions();

        try {
            // 开始事务
            kafkaProducer.beginTransaction();

            // 在事务内发送多条消息（可以发往不同Topic/Partition）
            kafkaProducer.send(new ProducerRecord<>("topic-a", "key1", "value1"));
            kafkaProducer.send(new ProducerRecord<>("topic-b", "key2", "value2"));
            kafkaProducer.send(new ProducerRecord<>("topic-a", "key3", "value3"));

            // ... 这里可以执行一些本地逻辑，例如数据库操作

            // 如果一切正常，提交事务
            kafkaProducer.commitTransaction();
            System.out.println("事务提交成功！");

        } catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
            // 这些是不可恢复的异常，必须关闭生产者
            kafkaProducer.close();
        } catch (KafkaException e) {
            // 对于其他可恢复的异常，中止当前事务
            System.out.println("事务失败，进行回滚: " + e);
            kafkaProducer.abortTransaction();
        }
    }


    @Test
    public void demoProducer() {
        Properties props = new Properties();
        //设置接入点
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, Constants.KAFKA_BOOTSTRAP_SERVERS);
        props.put("acks", "all");

        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        //Kafka消息的序列化方式。
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        //请求的最长等待时间。
        props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 30 * 1000);
        //设置客户端内部重试次数。
        props.put(ProducerConfig.RETRIES_CONFIG, 5);
        //设置客户端内部重试间隔。
        props.put(ProducerConfig.RECONNECT_BACKOFF_MS_CONFIG, 3000);

        //构造Producer对象，注意，该对象是线程安全的，一般来说，一个进程内一个Producer对象即可。
        //如果想提高性能，可以多构造几个对象，但不要太多，最好不要超过5个。
        Producer<String, String> producer = new KafkaProducer<String, String>(props);


        for (int i = 0; i < 100; i++) {
            try {
                Future<RecordMetadata> future = producer.send(
                        new ProducerRecord<String, String>(TOPIC, Integer.toString(i), "hello " + i));
                producer.flush();
                //同步获得Future对象的结果。
                RecordMetadata meta = future.get();
                System.out.println(meta.toString());
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                //客户端内部重试之后，仍然发送失败，业务要应对此类错误。
                e.printStackTrace();
            }

        }
        System.err.println("OVER");
        producer.close();
    }

}

